﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Web;

namespace RabbitMQ_Test2
{
    public class RabbitMQUtility : IDisposable
    {
        //public class QMessage
        //{
        //    public int Title { get; set; }
        //    public string value { get; set; }
        //    public DateTime dateTime { get; set; }
        //    public QMessage() { }
        //    public QMessage(DateTime dateTime, string value)
        //    {
        //        this.value = value;
        //        this.dateTime = dateTime;
        //    }

        //}

        //private IBasicProperties properties { get; set; }
        //private ConnectionFactory factory { get; set; }


        //private IConnection _SendConnection;
        //private readonly object Send_locker = new object();
        //private IConnection SendConnection
        //{
        //    get
        //    {
        //        if (_SendConnection == null)
        //        {
        //            lock (Send_locker)
        //            {
        //                if (_SendConnection == null)
        //                {
        //                    _SendConnection = factory.CreateConnection();
        //                }
        //            }
        //        }
        //        return _SendConnection;
        //    }
        //    set
        //    {
        //        _SendConnection = value;
        //    }
        //}

        //private IModel _SendChannel;

        //private IModel SendChannel
        //{
        //    get
        //    {
        //        if (_SendChannel == null)
        //        {
        //            lock (Send_locker)
        //            {
        //                if (_SendChannel == null)
        //                {
        //                    _SendChannel = this.SendConnection.CreateModel();
        //                    var channel = _SendChannel;
        //                    //设置交换器的类型
        //                    channel.ExchangeDeclare(exchangeName, exchangeType);
        //                    //声明一个队列，设置队列是否持久化，排他性，与自动删除
        //                    channel.QueueDeclare(queueName, true, false, false, null);
        //                    //绑定消息队列，交换器，routingkey
        //                    channel.QueueBind(queueName, exchangeName, routingKey);

        //                    //队列持久化
        //                    properties = channel.CreateBasicProperties();
        //                    properties.Persistent = true;
        //                    properties.DeliveryMode = deliveryMode;//硬盘序列化
        //                }
        //            }
        //        }
        //        return _SendChannel;
        //    }
        //    set
        //    {
        //        _SendChannel = value;
        //    }
        //}


        //private readonly object Pull_locker = new object();
        //private IConnection _PullConnection;

        //private IConnection PullConnection
        //{
        //    get
        //    {
        //        if (_PullConnection == null)
        //        {
        //            lock (Pull_locker)
        //            {
        //                if (_PullConnection == null)
        //                {
        //                    _PullConnection = factory.CreateConnection();
        //                }
        //            }
        //        }
        //        return _PullConnection;
        //    }
        //    set
        //    {
        //        _PullConnection = value;
        //    }
        //}

        //private IModel _PullChannel;

        //private IModel PullChannel
        //{
        //    get
        //    {
        //        if (_PullChannel == null)
        //        {
        //            lock (Pull_locker)
        //            {
        //                if (_PullChannel == null)
        //                {
        //                    _PullChannel = this.PullConnection.CreateModel();
        //                    var channel2 = this._PullChannel;
        //                    channel2.ExchangeDeclare(exchangeName, exchangeType);
        //                    channel2.QueueDeclare(this.queueName, true, false, false, null);
        //                    channel2.QueueBind(this.queueName, exchangeName, routingKey);

        //                }
        //            }
        //        }
        //        return _PullChannel;
        //    }
        //    set
        //    {
        //        _PullChannel = value;
        //    }
        //}
        //private QueueingBasicConsumer consumer;

        private string HostName { get; set; }
        private string UserName { get; set; }
        private string Password { get; set; }
        private ushort heartbeatTime { get; set; }
        private string VirtualHost { get; set; }
        private bool AutomaticRecoveryEnabled { get; set; }

        private string ip { get; set; }
        private ushort port { get; set; }
        private string queueName { get; set; }
        private string exchangeType { get; set; }
        private string exchangeName { get; set; }
        private string routingKey { get; set; }
        private byte deliveryMode { get; set; }

        private Uri uri { get; set; }
        public RabbitMQUtility(string queueName,
            string exchangeName,
            string ip,
            ushort port,
            string UserName,
            string Password,
            string HostName,
            string VirtualHost,
            string routingKey = "*",
            string exchangeType = "topic",
            bool AutomaticRecoveryEnabled = true,
            byte deliveryMode = 2,
            ushort heartbeatTime = 60)
        {
            this.queueName = queueName;
            this.exchangeType = exchangeType;
            this.exchangeName = exchangeName;
            this.deliveryMode = deliveryMode;
            this.routingKey = routingKey;
            var uri = new Uri("amqp://" + ip + ":" + port + "/");
            //factory = new ConnectionFactory
            //{
            this.HostName = HostName;
            this.UserName = UserName;
            this.Password = Password;
            this.heartbeatTime = heartbeatTime;
            this.uri = uri;
            this.VirtualHost = VirtualHost;
            this.AutomaticRecoveryEnabled = AutomaticRecoveryEnabled;
            this.routingKey = queueName;
            //};           

        }
        public void Dispose()
        {

        }
        public async Task<bool?> Producer(string value)
        {
            return await Task.Run(() =>
            {
                bool? returnValue = null;
                try
                {
                    ConnectionFactory factory = new ConnectionFactory
                    {
                        HostName = HostName,
                        UserName = UserName,
                        Password = Password,
                        RequestedHeartbeat = heartbeatTime,
                        Endpoint = new AmqpTcpEndpoint(uri),
                        VirtualHost = VirtualHost,
                        AutomaticRecoveryEnabled = AutomaticRecoveryEnabled
                    };
                    using (var connection = factory.CreateConnection())
                    {
                        using (var channel = connection.CreateModel())
                        {
                            //设置交换器的类型
                            channel.ExchangeDeclare(exchangeName, exchangeType);
                            //声明一个队列，设置队列是否持久化，排他性，与自动删除
                            channel.QueueDeclare(queueName, true, false, false, null);
                            //绑定消息队列，交换器，routingkey
                            channel.QueueBind(queueName, exchangeName, routingKey);

                            //队列持久化
                            IBasicProperties properties = channel.CreateBasicProperties();
                            properties.Persistent = true;
                            properties.DeliveryMode = deliveryMode;//硬盘序列化
                            //var m = new QMessage(DateTime.Now, value);
                            var body = Encoding.UTF8.GetBytes(value);
                            //发送信息
                            try
                            {
                                channel.BasicPublish(exchangeName, routingKey, properties, body);
                                returnValue = true;
                            }
                            catch (Exception oe)
                            {
                                string err = oe.Message + oe.StackTrace;
                            }

                        }
                    }
                }
                catch (Exception oe)
                {
                    string err = oe.Message + oe.StackTrace;
                }
                return returnValue;
            });
            //return await Task.Run(() =>
            //{
            //    bool? returnValue = null;
            //    try
            //    {
            //        byte deliveryMode = this.deliveryMode;//2 为硬盘序列化

            //        var channel = this.SendChannel;
            //        {
            //            ////设置交换器的类型
            //            //channel.ExchangeDeclare(exchangeName, exchangeType);
            //            ////声明一个队列，设置队列是否持久化，排他性，与自动删除
            //            //channel.QueueDeclare(queueName, true, false, false, null);
            //            ////绑定消息队列，交换器，routingkey
            //            //channel.QueueBind(queueName, exchangeName, routingKey);

            //            ////队列持久化
            //            //IBasicProperties properties = channel.CreateBasicProperties();
            //            //properties.Persistent = true;
            //            //properties.DeliveryMode = deliveryMode;//硬盘序列化
            //            //var m = new QMessage(DateTime.Now, value);
            //            //var body = Encoding.UTF8.GetBytes(Newtonsoft.Json.JsonConvert.SerializeObject(m));
            //            var body = Encoding.UTF8.GetBytes(value);
            //            //发送信息
            //            try
            //            {
            //                //lock(push_locker)
            //                //{
            //                    channel.BasicPublish(exchangeName, routingKey, properties, body);
            //                returnValue = true;
            //            }
            //            catch (Exception oe)
            //            {
            //                string err = oe.Message;
            //                try
            //                {
            //                    System.IO.File.AppendAllText(@"c:\log111.txt", err+System.Environment.NewLine);
            //                }
            //                catch (Exception oe2)
            //                {
            //                    err = oe.Message;
            //                }
            //            }

            //        }
            //    }
            //    catch (Exception oe)
            //    {
            //        string err = oe.Message;
            //        try
            //        {
            //            System.IO.File.AppendAllText(@"c:\log111.txt", err + System.Environment.NewLine);
            //        }
            //        catch (Exception oe2)
            //        {
            //            err = oe.Message;
            //        }
            //    }
            //    return returnValue;
            //});
        }

        public async Task<List<string>> Consumer(uint length)
        {
            return await Task.Run(async () =>
            {
                List<string> value = new List<string>();
                for (int i = 0; i < length; i++)
                {
                    string messageStr = await Consumer();
                    if (messageStr == null)
                    {
                        break;
                    }
                    if (!value.Contains(messageStr))
                    {
                        value.Add(messageStr);
                    }
                    else
                    {
                        i--;
                    }
                }
                //try
                //{
                //    var factory = new ConnectionFactory
                //    {
                //        HostName = HostName,
                //        UserName = UserName,
                //        Password = Password,
                //        RequestedHeartbeat = heartbeatTime,
                //        Endpoint = new AmqpTcpEndpoint(uri),
                //        VirtualHost = VirtualHost,
                //        AutomaticRecoveryEnabled = AutomaticRecoveryEnabled
                //    };
                //    using (var connection = factory.CreateConnection())
                //    {
                //        using (var channel = connection.CreateModel())
                //        {
                //            channel.ExchangeDeclare(exchangeName, exchangeType);
                //            channel.QueueDeclare(queueName, true, false, false, null);
                //            channel.QueueBind(queueName, exchangeName, routingKey);
                //            //定义这个队列的消费者
                //            QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                //            for (int i = 0; i < length; i++)
                //            {
                //                //false为手动应答，true为自动应答
                //                channel.BasicQos(0, 1, false);
                //                channel.BasicConsume(queueName, false, consumer);
                //                while (true)
                //                {
                //                    try
                //                    {
                //                        BasicDeliverEventArgs ea = null;
                //                        //QMessage message = null;
                //                        string messageStr = null;
                //                        try
                //                        {
                //                            ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
                //                            if (ea == null)
                //                            {
                //                                goto end;
                //                            }
                //                            byte[] bytes = ea.Body;
                //                            if (bytes == null)
                //                            {
                //                                goto end;
                //                            }
                //                            messageStr = Encoding.UTF8.GetString(bytes);
                //                            //message = Newtonsoft.Json.JsonConvert.DeserializeObject<QMessage>(messageStr);
                //                            if (!value.Contains(messageStr))
                //                            {
                //                                value.Add(messageStr);
                //                            }
                //                            else
                //                            {
                //                                i--;
                //                            }
                //                        }
                //                        catch (Exception oe)
                //                        {
                //                            string err = oe.Message + oe.StackTrace;
                //                            goto end;
                //                        }
                //                        //Console.WriteLine("Receive a Message, DateTime:" + message.dateTime.ToString("yyyy-MM-dd HH:mm:ss") + " Title:" + message.Title);
                //                        try
                //                        {
                //                            channel.BasicAck(ea.DeliveryTag, false);//这条应该是删除消息的语句 配合插入同内容流水id的结果队列的内容 应该就可以完成一个闭合  暂时没找到标记已经消费的。。。这个到时候要测试一下
                //                            ////如果是自动应答，下下面这句代码不用写啦。
                //                            //if ((Convert.ToInt32(message.Title) % 2) == 1)
                //                            //{
                //                            //channel.BasicAck(ea.DeliveryTag, false);
                //                            //}
                //                        }
                //                        catch (Exception oe)
                //                        {
                //                            string err = oe.Message + oe.StackTrace;
                //                            goto end;
                //                        }
                //                    }
                //                    catch (Exception oe)
                //                    {
                //                        string err = oe.Message + oe.StackTrace;
                //                        goto end;
                //                    }
                //                    break;
                //                }
                //                continue;
                //            end:
                //                break;
                //            }
                //        }
                //    }
                //}
                //catch (Exception oe)
                //{
                //    string err = oe.Message + oe.StackTrace;
                //}
                return value.ToList();
            });



        }
        public async Task<string> Consumer()
        {
            return await Task.Run(() =>
            {
                string value = null;
                try
                {
                    var factory = new ConnectionFactory
                    {
                        HostName = HostName,
                        UserName = UserName,
                        Password = Password,
                        RequestedHeartbeat = heartbeatTime,
                        Endpoint = new AmqpTcpEndpoint(uri),
                        VirtualHost = VirtualHost,
                        AutomaticRecoveryEnabled = AutomaticRecoveryEnabled
                    };
                    using (var connection = factory.CreateConnection())
                    {
                        using (var channel = connection.CreateModel())
                        {
                            channel.ExchangeDeclare(exchangeName, exchangeType);
                            channel.QueueDeclare(queueName, true, false, false, null);
                            channel.QueueBind(queueName, exchangeName, routingKey);
                            //定义这个队列的消费者
                            QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                            //false为手动应答，true为自动应答
                            channel.BasicQos(0, 1, false);
                            channel.BasicConsume(queueName, false, consumer);
                            while (true)
                            {
                                try
                                {
                                    BasicDeliverEventArgs ea = null;
                                    //QMessage message = null;
                                    string messageStr = null;
                                    try
                                    {

                                        consumer.Queue.Dequeue(1000, out ea);
                                        if (ea != null)
                                        {
                                            byte[] bytes = ea.Body;
                                            messageStr = Encoding.UTF8.GetString(bytes);
                                            //message = Newtonsoft.Json.JsonConvert.DeserializeObject<QMessage>(messageStr);
                                            value = messageStr;
                                        }
                                    }
                                    catch (Exception oe)
                                    {
                                        string err = oe.Message + oe.StackTrace;
                                    }
                                    //Console.WriteLine("Receive a Message, DateTime:" + message.dateTime.ToString("yyyy-MM-dd HH:mm:ss") + " Title:" + message.Title);
                                    try
                                    {
                                        if (ea != null)
                                        {
                                            channel.BasicAck(ea.DeliveryTag, false);//这条应该是删除消息的语句 配合插入同内容流水id的结果队列的内容 应该就可以完成一个闭合  暂时没找到标记已经消费的。。。这个到时候要测试一下
                                            ////如果是自动应答，下下面这句代码不用写啦。
                                            //if ((Convert.ToInt32(message.Title) % 2) == 1)
                                            //{
                                            //channel.BasicAck(ea.DeliveryTag, false);
                                            //}
                                        }
                                    }
                                    catch (Exception oe)
                                    {
                                        string err = oe.Message + oe.StackTrace;
                                    }
                                }
                                catch (Exception oe)
                                {
                                    string err = oe.Message + oe.StackTrace;
                                }
                                break;
                            }
                        }
                    }
                }
                catch (Exception oe)
                {
                    string err = oe.Message + oe.StackTrace;
                }
                return value;
            });
            //return await Task.Run(() =>
            //{
            //    string value = null;
            //    try
            //    {
            //        //var qName = this.queueName;
            //        //var exchangeName = this.exchangeName;
            //        //var exchangeType = this.exchangeType;//topic、fanout
            //        //var routingKey = this.routingKey;

            //        var channel = this.PullChannel;
            //        {
            //            //定义这个队列的消费者
            //            consumer = new QueueingBasicConsumer(channel);
            //            //false为手动应答，true为自动应答
            //            channel.BasicQos(0, 1, false);
            //            channel.BasicConsume(queueName, false, consumer);

            //            //false为手动应答，true为自动应答
            //            try
            //            {
            //                BasicDeliverEventArgs ea = null;
            //                //QMessage message = null;
            //                string messageStr = null;
            //                try
            //                {
            //                    ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
            //                    byte[] bytes = ea.Body;
            //                    messageStr = Encoding.UTF8.GetString(bytes);
            //                    //message = Newtonsoft.Json.JsonConvert.DeserializeObject<QMessage>(messageStr);
            //                    value = messageStr;
            //                }
            //                catch (Exception oe)
            //                {
            //                    string err = oe.Message;
            //                }
            //                //Console.WriteLine("Receive a Message, DateTime:" + message.dateTime.ToString("yyyy-MM-dd HH:mm:ss") + " Title:" + message.Title);
            //                try
            //                {
            //                    channel.BasicAck(ea.DeliveryTag, false);//这条应该是删除消息的语句 配合插入同内容流水id的结果队列的内容 应该就可以完成一个闭合  暂时没找到标记已经消费的。。。这个到时候要测试一下
            //                    ////如果是自动应答，下下面这句代码不用写啦。
            //                    //if ((Convert.ToInt32(message.Title) % 2) == 1)
            //                    //{
            //                    //channel.BasicAck(ea.DeliveryTag, false);
            //                    //}
            //                }
            //                catch (Exception oe)
            //                {
            //                    string err = oe.Message;
            //                }
            //            }
            //            catch (Exception oe)
            //            {
            //                string err = oe.Message;
            //            }
            //        }
            //    }
            //    catch (Exception oe)
            //    {
            //        string err = oe.Message;
            //    }
            //    return value;
            //});
        }
        //public static void consumer_Received(object sender, BasicDeliverEventArgs e)
        //{
        //channel.ExchangeDeclare(exchangeName, exchangeType);
        //channel.QueueDeclare(qName, true, false, false, null);
        //channel.QueueBind(qName, exchangeName, routingKey);
        ////定义这个队列的消费者
        //QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
        ////false为手动应答，true为自动应答
        //channel.BasicConsume(qName, false, consumer);
        //while (true)
        //{
        //    BasicDeliverEventArgs ea = null;
        //    QMessage message = null;
        //    string messageStr = null;
        //    try
        //    {
        //        ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
        //        byte[] bytes = ea.Body;
        //        messageStr = Encoding.UTF8.GetString(bytes);
        //        message = Newtonsoft.Json.JsonConvert.DeserializeObject<QMessage>(messageStr);
        //        value = messageStr;
        //    }
        //    catch { }
        //    Console.WriteLine("Receive a Message, DateTime:" + message.dateTime.ToString("yyyy-MM-dd HH:mm:ss") + " Title:" + message.Title);
        //    try
        //    {
        //        //如果是自动应答，下下面这句代码不用写啦。
        //        if ((Convert.ToInt32(message.Title) % 2) == 1)
        //        {
        //            channel.BasicAck(ea.DeliveryTag, false);
        //        }
        //    }
        //    catch { }
        //    break;
        //}
        //}
    }
}